package com.at.sink10;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;

/**
 * @author huangchao E-mail:fengquan8866@163.com
 * @version 创建时间：2024/9/25 20:34
 */
public class SinkCustom5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> sensorDS = env.socketTextStream("master", 7777);

        sensorDS.addSink(new MySink());

        env.execute();
    }

    public static class MySink extends RichSinkFunction<String> {
        private static final long serialVersionUID = 1513378032005985337L;
        Connection conn = null;
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            // 创建连接
        }

        @Override
        public void close() throws Exception {
            super.close();
            // 销毁、清理
        }

        /**
         * sink的核心逻辑，写出的逻辑就写在这个方法里
         * @param value The input record.
         * @param context Additional context about the input record.
         * @throws Exception
         */
        @Override
        public void invoke(String value, Context context) throws Exception {
            // 写出逻辑
            // 这个方法是 来一条数据，调用一次，所以不要在这里创建 连接对象
        }
    }
}
